package com.wjk.kylin.mall.product.es;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.wjk.kylin.mall.common.core.constant.Constants;
import com.wjk.kylin.mall.product.constant.ProductConstants;
import com.wjk.kylin.mall.product.entity.es.ProductESBean;
import com.wjk.kylin.mall.product.utils.EsClientUtil;
import com.wjk.kylin.mall.product.utils.EsIterator;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;


/**
 * @Author: wangjinkui
 * @Date: 2021/6/30
 * @Description : 商品 es
 * 后续改为 通过监听 binlog -> canal -> kafka -> 消费者服务 -> es
 */
@Slf4j
@Service
public class ProductEsServiceImpl {
    @Autowired
    private RestHighLevelClient client;
    @Autowired
    private EsClientUtil esClientUtil;

    /**
     * @param existProductIdList
     * @return
     */
    public Map<Long, ProductESBean> getProductEsMap(List<Long> existProductIdList) {
        List<ProductESBean> list = new ArrayList<>();
        //超过1024
        List<Long> idList = existProductIdList.stream().distinct().collect(Collectors.toList());
        if (idList.size() > Constants.ES_MAX_PARAMS_NUM) {
            List<List<Long>> partitionList = Lists.partition(idList, Constants.ES_MAX_PARAMS_NUM);
            for (List<Long> partition : partitionList) {
                getProductEsList(list, partition);
            }
        } else {
            getProductEsList(list, idList);
        }

        return list.stream().collect(Collectors.toMap(ProductESBean::getId, Function.identity(), (key1, key2) -> key2));
    }

    /**
     * 新增或更新、删除
     *
     * @param insertOrUpdateEsList
     * @param deleteIdList
     */
    public void bulkProductEsList(List<ProductESBean> insertOrUpdateEsList, List<Long> deleteIdList) {
        if (CollectionUtil.isNotEmpty(insertOrUpdateEsList)) {
            try {
                //支持新增更新
                BulkRequest bulkRequest = new BulkRequest();
                for (ProductESBean entity : insertOrUpdateEsList) {
                    IndexRequest indexRequest = new IndexRequest(ProductConstants.PRODUCT_ES_INDEX);
                    indexRequest.id(entity.getId().toString()).source(JSON.toJSONString(entity), XContentType.JSON);
                    bulkRequest.add(indexRequest);
                }
                BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
                log.info("新增或更新状态：{}", response.status());
            } catch (Exception e) {
                log.error("es insert or update error:", e);
                throw new RuntimeException(e);
            }
        }
        if (CollectionUtil.isNotEmpty(deleteIdList)) {
            try {
                // 批量删除数据
                BulkRequest request = new BulkRequest();
                deleteIdList.forEach(s -> request.add(new DeleteRequest().index(ProductConstants.PRODUCT_ES_INDEX).id(s.toString())));

                BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);
                log.info("删除状态：{}", response.status());
            } catch (Exception e) {
                log.error("es delete error:", e);
                throw new RuntimeException(e);
            }
        }
    }


    /**
     * 迭代器查询
     *
     * @param list
     * @param partition
     */
    private void getProductEsList(List<ProductESBean> list, List<Long> partition) {
        SearchSourceBuilder builder = new SearchSourceBuilder();
        builder.query(QueryBuilders.termsQuery("id", partition));
        builder.size(Constants.ES_QUERY_PAGE_SIZE);//2000

        EsIterator<ProductESBean> iterator = esClientUtil.selectEsIterator(ProductConstants.PRODUCT_ES_INDEX, 2L, builder, ProductESBean.class);

        while (iterator.hasNext()) {
            list.add(iterator.next());
        }
    }
}
